/**
 * Software License, Version 1.0
 * 
 * Copyright 2003 The Trustees of Indiana University.  All rights reserved.
 * 
 *
 *Redistribution and use in source and binary forms, with or without 
 *modification, are permitted provided that the following conditions are met:
 *
 *1) All redistributions of source code must retain the above copyright notice,
 * the list of authors in the original source code, this list of conditions and
 * the disclaimer listed in this license;
 *2) All redistributions in binary form must reproduce the above copyright 
 * notice, this list of conditions and the disclaimer listed in this license in
 * the documentation and/or other materials provided with the distribution;
 *3) Any documentation included with all redistributions must include the 
 * following acknowledgement:
 *
 *"This product includes software developed by the Community Grids Lab. For 
 * further information contact the Community Grids Lab at 
 * http://communitygrids.iu.edu/."
 *
 * Alternatively, this acknowledgement may appear in the software itself, and 
 * wherever such third-party acknowledgments normally appear.
 * 
 *4) The name Indiana University or Community Grids Lab or NaradaBrokering, 
 * shall not be used to endorse or promote products derived from this software 
 * without prior written permission from Indiana University.  For written 
 * permission, please contact the Advanced Research and Technology Institute 
 * ("ARTI") at 351 West 10th Street, Indianapolis, Indiana 46202.
 *5) Products derived from this software may not be called NaradaBrokering, 
 * nor may Indiana University or Community Grids Lab or NaradaBrokering appear
 * in their name, without prior written permission of ARTI.
 * 
 *
 * Indiana University provides no reassurances that the source code provided 
 * does not infringe the patent or any other intellectual property rights of 
 * any other entity.  Indiana University disclaims any liability to any 
 * recipient for claims brought by any other entity based on infringement of 
 * intellectual property rights or otherwise.  
 *
 *LICENSEE UNDERSTANDS THAT SOFTWARE IS PROVIDED "AS IS" FOR WHICH NO 
 *WARRANTIES AS TO CAPABILITIES OR ACCURACY ARE MADE. INDIANA UNIVERSITY GIVES
 *NO WARRANTIES AND MAKES NO REPRESENTATION THAT SOFTWARE IS FREE OF 
 *INFRINGEMENT OF THIRD PARTY PATENT, COPYRIGHT, OR OTHER PROPRIETARY RIGHTS. 
 *INDIANA UNIVERSITY MAKES NO WARRANTIES THAT SOFTWARE IS FREE FROM "BUGS", 
 *"VIRUSES", "TROJAN HORSES", "TRAP DOORS", "WORMS", OR OTHER HARMFUL CODE.  
 *LICENSEE ASSUMES THE ENTIRE RISK AS TO THE PERFORMANCE OF SOFTWARE AND/OR 
 *ASSOCIATED MATERIALS, AND TO THE PERFORMANCE AND VALIDITY OF INFORMATION 
 *GENERATED USING SOFTWARE.
 */
package cgl.narada.matching.tagvalue;

import java.util.Enumeration;
import java.util.Hashtable;
import java.util.StringTokenizer;
import java.util.Vector;

import cgl.narada.matching.EventDestinations;
import cgl.narada.protocol.Connection;
import cgl.narada.protocol.Destinations;
import cgl.narada.protocol.Gateway;
import cgl.narada.protocol.GatewayInfo;
import cgl.narada.protocol.NodeAddress;
import cgl.narada.protocol.ProtocolHandler;
import cgl.narada.protocol.ProtocolIDFactory;
import cgl.narada.protocol.TagValueProfilePropagation;

/**
 * This class implements the matching algorithm. This provides methods to add a
 * subscription and also to remove interest in a subscription predicate. The
 * matching algorithm should also be able to compute destinations based on the
 * general matching algorithm.
 * 
 * @author Shrideep Pallickara $Revision$ $Date$
 */

public class ClientMatching implements MatchingTree, ClientMatchingDebugFlags {
    /** This Hashtable maintains the list of different kinds of first tags */
    private Hashtable firstTags;

    /** Profile Propagation Handler */
    private TagValueProfilePropagation tvProtocol;

    /** The unit level for this matching tree */
    private int unitLevel;

    /** The system level */
    private int systemLevel;

    /** The destination for this node */
    Destinations thisDestination;

    private Hashtable destinationsInfo;

    private Hashtable predicates;

    /** The Matching constructor */
    public ClientMatching(int unitLevel, int systemLevel,
        TagValueProfilePropagation tvProtocol) {
        this.unitLevel = unitLevel;
        this.systemLevel = systemLevel;
        this.tvProtocol = tvProtocol;
        firstTags = new Hashtable();

        destinationsInfo = new Hashtable();
        predicates = new Hashtable();
    }

    /**
     * This method returns the level assocaited with this matching tree. A node
     * could be a gateway at multiple levels. How it handles destinations at
     * different levels, and propagates destinations at different levels is
     * dicatated by the level of the matching tree which computes the
     * destinations and initiates profile propagation changes.
     * 
     * @return The level of the matching tree.
     */
    public int getMatchingLevel() {
        return unitLevel;
    }

    /** This method sets the destination address for this node */
    public void setNodeDestination(Destinations destination) {
        thisDestination = destination;
    }

    /**
     * This method adds a subscription predicate to the matching tree
     * 
     * @param predicate
     *            The predicate comprising the subscription and the destination
     *            assocaited with the subscription.
     */
    public void addSubscriptionPredicate(Predicate predicate) {
        String predicateId = predicate.getPredicateId();
        if (predicateId != null) {
            if (predicates.containsKey(predicateId)) {
                /** This predicate has of course been processed before. */
                return;
            }
        }

        if (ClientMatching_Debug) {
            System.out.println("Adding predicate " + predicate);
        }

        Vector tokenVector = processSubscription(predicate.getSubscription());
        if (tokenVector == null) {
            System.out.println("Error in subscription predicate ... returning");
            return;
        }

        int size = tokenVector.size();
        String firstTag = (String) tokenVector.firstElement();
        ClientMatchingTreeNode tagNode;
        if (firstTags.containsKey(firstTag)) {
            tagNode = (ClientMatchingTreeNode) firstTags.get(firstTag);
        } else {
            tagNode = new ClientMatchingTreeNode(firstTag);
            firstTags.put(firstTag, tagNode);
        }
        int destination = predicate.getDestination();
        ClientMatchingTreeEdge edge;
        ClientMatchingTreeNode node;
        node = tagNode;
        for (int i = 1; i < size - 1;) {
            edge = tagNode.createEdgeLeadingIntoNode((String) tokenVector
                .elementAt(i++), (String) tokenVector.elementAt(i++),
                                                     destination);
            tagNode = edge.getLeadingIntoNode();
        }
        boolean propagate = tagNode
            .createEdgeWithDestinations((String) tokenVector.lastElement(),
                                        destination);

        if (predicateId != null) {
            Integer destKey = new Integer(destination);
            if (!destinationsInfo.containsKey(destKey)) {
                destinationsInfo.put(destKey, new Hashtable());
            }

            ((Hashtable) destinationsInfo.get(destKey)).put(predicateId,
                                                            predicate);
            predicates.put(predicateId, predicate);

        } /* end (predicateId!=null) */

        if (propagate) {
            if (ClientMatching_Debug) {
                System.out.println("This predicate needs to be propagated ");
            }
        }
    }

    public synchronized void removePredicatesForDestination(
                                                            Object clientIdentifier) {
        if (!destinationsInfo.containsKey(clientIdentifier)) {
            if (ClientMatching_Debug) {
                System.out.println("ClientMatching: Not aware of destination ["
                    + clientIdentifier
                    + "] submitted for removal from matching tree");
            }
            return;
        }

        Hashtable predicateIdsToRemove = (Hashtable) destinationsInfo
            .get(clientIdentifier);
        Enumeration e = predicateIdsToRemove.keys();
        while (e.hasMoreElements()) {
            Object _removeId = e.nextElement();
            Predicate _predicate = (Predicate) predicateIdsToRemove
                .remove(_removeId);
            removeSubscriptionPredicate(_predicate);
        }

    }

    /**
     * This method removes a subscription predicate to the matching tree
     * 
     * @param predicate
     *            The predicate comprising the subscription and the destination
     *            assocaited with the subscription.
     */
    public synchronized void removeSubscriptionPredicate(Predicate predicate) {
        String predicateId = predicate.getPredicateId();
        if (predicateId != null) {
            if (!predicates.containsKey(predicateId)) {
                /** This predicate has of course been processed before. */
                System.out
                    .println("Trying to remove a non-existent predicateId");
                return;
            }
        }

        Vector tokenVector = processSubscription(predicate.getSubscription());
        if (tokenVector == null) {
            System.out.println("Error in subscription predicate ... returning");
            return;
        }
        int size = tokenVector.size();
        String firstTag = (String) tokenVector.firstElement();
        /**
         * Check to see if the first tag exists, if it doesn't we are trying to
         * remove a predicate that does not exist
         */
        if (!firstTags.containsKey(firstTag)) {
            System.out.println("Trying to remove a non-existant predicate ");
            return;
        }
        int destination = predicate.getDestination();
        ClientMatchingTreeEdge edge;
        ClientMatchingTreeNode node;
        node = (ClientMatchingTreeNode) firstTags.get(firstTag);
        for (int i = 1; i < size - 1;) {
            edge = node.removeEdgeLeadingIntoNode((String) tokenVector
                .elementAt(i++), (String) tokenVector.elementAt(i++),
                                                  destination);
            if (edge == null) {
                System.out.println("ClientMatching:removeSubPredicate::"
                    + "All edges and nodes from this point"
                    + " on have been discarded");

                /**
                 * This change needs to be propagated since it has resulted in
                 * the removal of an edge
                 */
                if (predicateId != null) {
                    Object _clientIdentifier = new Integer(destination);
                    ((Hashtable) destinationsInfo.get(_clientIdentifier))
                        .remove(predicateId);
                    if (((Hashtable) destinationsInfo.get(_clientIdentifier))
                        .size() == 0) {
                        destinationsInfo.remove(_clientIdentifier);
                        System.out.println("ClientMatching: Destination ["
                            + _clientIdentifier
                            + "] removed from DestinationsInfo Table");
                    }
                    predicates.remove(predicateId);
                } /* if (predicateId != null) */

                return;
            } /* end if (edge == null) */
            node = edge.getLeadingIntoNode();
        }

        boolean propagate = node
            .removeEdgeWithDestinations((String) tokenVector.lastElement(),
                                        destination);
        if (predicateId != null) {
            Object _clientIdentifier = new Integer(destination);
            ((Hashtable) destinationsInfo.get(_clientIdentifier))
                .remove(predicateId);
            if (((Hashtable) destinationsInfo.get(_clientIdentifier)).size() == 0) {
                destinationsInfo.remove(_clientIdentifier);
                System.out.println("ClientMatching: Destination ["
                    + _clientIdentifier + "] has been removed "
                    + "from the DestinationsInfo Table");
            }
            predicates.remove(predicateId);
            System.out.println("Removed Predicate ID" + predicateId);
        } /* if (predicateId != null) */

        if (propagate) {
            System.out.println("This change needs to be propagated ");
        }

    }

    public void propagateChangesToHigherLevels(Predicate predicate,
                                               boolean addPredicate) {
        /**
         * The first step ofcourse is to replace lower level destinations with
         * the destination ID at the next higher level
         */
        if (ClientMatching_Debug) {
            System.out.println("The destination associated with predicate "
                + predicate.getSubscription() + " is "
                + predicate.getDestination());
        }
        int temp = unitLevel;
        int tempDest;
        for (; temp < systemLevel;) {
            tempDest = thisDestination.getDestinationsAtLevel(temp);
            if (ClientMatching_Debug) {
                System.out.print("This destination would be replaced by "
                    + tempDest);
            }
            temp++;
            if (ClientMatching_Debug) {
                System.out.println(" for representation in a level(" + temp
                    + ") tree ");
            }
            tvProtocol.propagateProfileChange(new Predicate(predicate
                .getSubscription(), tempDest), temp, addPredicate);
        }
    }

    /**
     * This method is responsible for processing a subscription, and first
     * splitting them up into individual tag value pairs followed by splitting
     * these tag-value pairs into individual tags and pairs. The subscription
     * MUST comprise of tag-value pairs else this method will throw an error
     */
    private Vector processSubscription(String subscription) {
        if (ClientMatching_Debug)
            System.out.println("ClientMatching:: Processing subscription "
                + subscription);
        Vector tokenVector = new Vector();
        StringTokenizer tok = new StringTokenizer(subscription, ",");
        int i = 0;
        String tag;
        String value;
        while (tok.hasMoreTokens()) {
            i++;
            String tagValuePair = tok.nextToken();
            if (ClientMatching_Debug)
                System.out.println("TagValue Pair " + i + " ->" + tagValuePair);
            StringTokenizer tok2 = new StringTokenizer(tagValuePair, "=");
            if (tok2.countTokens() == 2) {
                tag = tok2.nextToken().trim();
                value = tok2.nextToken().trim();
                if (ClientMatching_Debug)
                    System.out.println("Tag = " + tag + " Value = " + value);
                tokenVector.addElement(tag);
                tokenVector.addElement(value);
            } else {
                System.out
                    .println("Malformed TAG=VALUE Pair in predicate with "
                        + "subscription " + subscription);
                return null;
            }
        }
        return tokenVector;
    }

    /**
     * This method return the destinations assocaited with an event that needs
     * to be matched to the matching tree
     * 
     * @return The destination as an INT
     */
    public Hashtable matchEvent(MEvent e) {
        String tag = e.getNextTag();
        if (tag == null) {
            System.out.println("ClientMatching::"
                + "Event doesn't have any tag/value pairs");
            return null;
        }

        if (ClientMatching_Debug) {
            System.out.println("Event to be matched is " + e);
        }
        if (firstTags.containsKey(tag)) {
            ClientMatchingTreeNode node = (ClientMatchingTreeNode) firstTags
                .get(tag);
            return node.matchEvent(e, new Hashtable());
        }
        if (ClientMatching_Debug)
            System.out
                .println("ClientMatching::Event doesn't have any matching "
                    + "destinations");
        return null;
    }

    /**
     * This method return the destinations assocaited with an event that needs
     * to be matched to the matching tree. The return value provides us with
     * updates that can be used for writing the event to stable storage.
     * 
     */
    public void matchEventForStorage(MEvent e, EventDestinations eDest) {
        e.resetLocalPointer();
        String tag = e.getNextTag();
        if (tag == null) {
            System.out.println("ClientMatching::Storage ->"
                + "Event doesn't have any tag/value pairs");
            return;
        }
        if (ClientMatching_Debug)
            System.out.println("MTag = " + tag);
        if (firstTags.containsKey(tag)) {
            ClientMatchingTreeNode node = (ClientMatchingTreeNode) firstTags
                .get(tag);
            node.matchEventForStorage(e, eDest);
        }
        System.out.println("ClientMatching::Storage ->"
            + "Event doesn't have any matching destinations");
    }

    public static void main(String[] args) {
        int[] address = { 5, 8, 10, 6 };
        NodeAddress nodeAddress = new NodeAddress(address);
        Destinations destinations = new Destinations(nodeAddress
            .getAddressInBytes());

        /** Building the connectivity graph */
        int[] a6 = { 6 };
        int[] a4 = { 2 };
        int[] a5 = { 3 };
        int[] a7 = { 4 };

        int[] aa = { 7 };
        int[] ab = { 8 };
        int[] ac = { 9 };
        int[] aSC_2 = { 10 };
        int[] aSC_3 = { 11 };
        int[] aSSC_B = { 20 };
        int[] aSSC_C = { 21 };
        int[] aSSC_D = { 22 };
        NodeAddress na6, na5, na4, na7, naa, nab, naSC_3, naSC_2, naSSC_B, naSSC_C, naSSC_D;
        na6 = new NodeAddress(a6);
        na5 = new NodeAddress(a5);
        na4 = new NodeAddress(a4);
        na7 = new NodeAddress(a7);

        naa = new NodeAddress(aa);
        nab = new NodeAddress(ab);

        naSC_2 = new NodeAddress(aSC_2);
        naSC_3 = new NodeAddress(aSC_3);

        naSSC_B = new NodeAddress(aSSC_B);
        naSSC_C = new NodeAddress(aSSC_C);
        naSSC_D = new NodeAddress(aSSC_D);

        int[] addressOfNode = { 5, 8, 10, 6 };
        NodeAddress thisNodeAddress = new NodeAddress(addressOfNode);
        ProtocolIDFactory idFactory = new ProtocolIDFactory(thisNodeAddress,
                                                            (short) 0);

        Connection c64 = new Connection(thisNodeAddress, na6, 0, na4, 0, 0,
                                        idFactory.getProtocolID());
        Connection c65 = new Connection(thisNodeAddress, na6, 0, na5, 0, 0,
                                        idFactory.getProtocolID());
        Connection c67 = new Connection(thisNodeAddress, na7, 0, na6, 0, 0,
                                        idFactory.getProtocolID());

        Connection c4a = new Connection(thisNodeAddress, na4, 0, naa, 1, 1,
                                        idFactory.getProtocolID());
        Connection cab = new Connection(thisNodeAddress, naa, 1, nab, 1, 1,
                                        idFactory.getProtocolID());

        Connection cbSC_3 = new Connection(thisNodeAddress, nab, 1, naSC_3, 2,
                                           2, idFactory.getProtocolID());
        Connection c5SC_2 = new Connection(thisNodeAddress, na5, 0, naSC_2, 2,
                                           2, idFactory.getProtocolID());
        Connection cSC_3SC_2 = new Connection(thisNodeAddress, naSC_2, 2,
                                              naSC_3, 2, 2, idFactory
                                                  .getProtocolID());

        Connection caSSC_B = new Connection(thisNodeAddress, naa, 1, naSSC_B,
                                            3, 3, idFactory.getProtocolID());
        Connection cSC_3SSC_B = new Connection(thisNodeAddress, naSC_3, 2,
                                               naSSC_B, 3, 3, idFactory
                                                   .getProtocolID());
        Connection cSSC_BSSC_D = new Connection(thisNodeAddress, naSSC_B, 3,
                                                naSSC_D, 3, 3, idFactory
                                                    .getProtocolID());
        Connection cSSC_CSSC_D = new Connection(thisNodeAddress, naSSC_C, 3,
                                                naSSC_D, 3, 3, idFactory
                                                    .getProtocolID());
        Connection cSC_2SSC_C = new Connection(thisNodeAddress, naSC_2, 2,
                                               naSSC_C, 3, 3, idFactory
                                                   .getProtocolID());

        Connection c7SC_3 = new Connection(thisNodeAddress, na7, 0, naSC_3, 2,
                                           2, idFactory.getProtocolID());

        GatewayInfo gatewayInfo = new GatewayInfo(3);
        gatewayInfo.setNodeAddress(thisNodeAddress);
        gatewayInfo.setVertexNode(new Gateway(na6, (short) 0, true));

        gatewayInfo.addConnection(c64);
        gatewayInfo.addConnection(c65);
        gatewayInfo.addConnection(c67);
        gatewayInfo.addConnection(c4a);
        gatewayInfo.addConnection(cab);
        gatewayInfo.addConnection(cbSC_3);
        gatewayInfo.addConnection(c5SC_2);
        gatewayInfo.addConnection(cSC_3SC_2);
        gatewayInfo.addConnection(caSSC_B);
        gatewayInfo.addConnection(cSC_3SSC_B);
        gatewayInfo.addConnection(cSSC_BSSC_D);
        gatewayInfo.addConnection(cSSC_CSSC_D);
        gatewayInfo.addConnection(cSC_2SSC_C);

        TagValueProfilePropagation ppp = new TagValueProfilePropagation(
                                                                        gatewayInfo,
                                                                        new ProtocolHandler());

        ClientMatching matching = new ClientMatching(0, 3, ppp);
        ppp.setDestination(destinations);
        matching.setNodeDestination(destinations);
        String sub1 = "A=a,B=b,C=c";
        String sub2 = "A=a, B=d, C=c";
        String sub3 = "A=a, B=*, C=c";
        String sub4 = "A=a, B=g, C=f, D=g";
        String sub5 = "Cool= Dude";
        matching.addSubscriptionPredicate(new Predicate(sub1, 1024));
        matching.addSubscriptionPredicate(new Predicate(sub2, 999));
        matching.addSubscriptionPredicate(new Predicate(sub3, 891));
        matching.addSubscriptionPredicate(new Predicate(sub3, 451));
        matching.addSubscriptionPredicate(new Predicate(sub4, 983));

        System.out.println("\n" + "Testing matching functions here ");
        MEvent event = new MEvent(sub1 + ",MesgNbr=1", 30);
        Hashtable dest = matching.matchEvent(event);
        if (dest == null) {
            System.out.println("Event has no matching destinations ");
        } else {
            Enumeration e = dest.keys();
            while (e.hasMoreElements()) {
                System.out.println(((Integer) e.nextElement()).intValue());
            }
        }

        matching.removeSubscriptionPredicate(new Predicate(sub4, 983));
    }
}
